/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package homomorphism.NewAPI;

import homomorphism.GroupLongPairComparator;
import homomorphism.HomomorphismBase;
//import homomorphism.ListPartitioner;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.ToolRunner;

import utility.LongPair;
import homomorphism.list.ThirdHomomorphismWrapper;
import homomorphism.TotalLongPairKeyComparator;

/**
 *
 * @author liuyu
 */
public class HomomorphismWrapperFor3rdThm<InVal, OtVal> extends HomomorphismBase {

    private static final String MyUsage = "usage:  [input file path]  [output file path] <-M [:num]> <-R [:num]>-h  :[usage]";
    private static final Log LOG = LogFactory.getLog(HomomorphismWrapperFor3rdThm.class);

    public HomomorphismWrapperFor3rdThm() {
    }

    public OtVal fold(ArrayList<InVal> values) throws Exception {
        throw new RuntimeException("Compiled Code");
    }

    public ArrayList<InVal> unfold(OtVal value) throws Exception {
        throw new RuntimeException("Compiled Code");
    }

    public static class MyFirstMapper<InVal, OtVal> extends ListIndexingMapper<InVal, OtVal> {
    }

    public static class MyFirstReducer<InVal, OtVal> extends ReducerFor3rdLstHomoThm1stPh<InVal, OtVal> {
    }

    public static class MySecondMapper<OtVal, InVal> extends MapperFor3rdLstHomoThm2ndPh<OtVal, InVal> {
    }

    public static class MySecondReducer<InVal, OtVal> extends ReducerFor3rdLstHomoThm2ndP<InVal, OtVal> {
    }

    public int run(String[] args) throws Exception {
        if (args.length < 3) {
            ThirdHomomorphismWrapper.printUsage(this, MyUsage);
            return -1;
        }
        Configuration conf = getConf();
        JobClient client1 = new JobClient(conf);
        ClusterStatus cluster = client1.getClusterStatus();

        /*
        int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
        String sort_reduces = conf.get(REDUCES_PER_HOST);
        if (sort_reduces != null) {
        num_reduces = cluster.getTaskTrackers() *
        Integer.parseInt(sort_reduces);
        }
         */

        Job job1 = new Job(conf);

        //  JobConf job1 = new JobConf(conf, this.getClass());
        job1.setJobName("First_phase!");
        job1.setJarByClass(this.getClass());

        job1 = initJobConf1(job1, MyFirstMapper.class, MyFirstReducer.class, args);
        if (job1 == null) {
            LOG.error(" job1 init  failed ");
            return -1;
        }
        Job job2 = new Job(conf);
        job2.setJobName("Second_phase!");
        job2.setJarByClass(this.getClass());
        job2 = initJobConf2(job2, MySecondMapper.class, MySecondReducer.class, args);
        // LOG.info("job generating finished!");

//        Date startTime = new Date();
//        System.out.println("The 1st Job started: " + startTime);

        int ret = job1.waitForCompletion(true) ? 0 : 1;

        if (ret == 0) {
            ret = job2.waitForCompletion(true) ? 0 : 1;
        }
       long time = job2.getFinishTime() -  job1.getStartTime();

       System.out.println(" All Jobs ended, The 2 jobs took: " + time/1000+ " seconds.");

//        Date end_time = new Date();
//        System.out.println("All Jobs ended: " + end_time);
//        System.out.println("The 2 jobs took "
//                + (end_time.getTime() - startTime.getTime()) / 1000 + " seconds.");

        return ret;
    }

//            public void SetFoldUnfoldArgumentsType(Class<?> userclass, JobConf job) throws Exception {
//
//        ParameterizedType aguType = (ParameterizedType) userclass.getGenericInterfaces()[0];
//
//        if (aguType.getActualTypeArguments().length != 2) {
//            //  throw new Exception(this.getClass().getSimpleName() + "does not implement right interface!\n");
//        }
//        //(T1)
//        Class lFoldingAguClass = aguType.getActualTypeArguments()[0].getClass();
//        Class lUnFoldingAguClass = aguType.getActualTypeArguments()[1].getClass();
//        job.setMapOutputValueClass(lUnFoldingAguClass);
//        //T2
//        job.setOutputValueClass(lFoldingAguClass);
//    }
    public Job initJobConf1(Job job,
            Class UserFirstMapperClass,
            Class UserFirstReducerClass,
            String[] args) throws Exception {

        Class<? extends InputFormat> inputFormatClass =
                SequenceFileInputFormat.class;
        Class<? extends OutputFormat> outputFormatClass =
                SequenceFileOutputFormat.class;

        // FileInputFormat.addInputPath(job,new Path(args[0]));
        FileInputFormat.setInputPaths(job, args[0]);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        JobClient client = new JobClient(job.getConfiguration());
        ClusterStatus cluster = client.getClusterStatus();

        int num_reducers = (int) (cluster.getMaxReduceTasks() * 0.9);
        if (num_reducers < 1) {
            num_reducers = 1;
        }

        String sort_reduces = job.getConfiguration().get(REDUCES_PER_HOST);
        if (sort_reduces != null) {
            num_reducers = cluster.getTaskTrackers()
                    * Integer.parseInt(sort_reduces);
        }
        int idx = ThirdHomomorphismWrapper.HasParameter("-R", args);
        if (idx >= 0 && args.length > idx) {
            num_reducers= Integer.parseInt(args[idx + 1]);
        }
         job.setNumReduceTasks(num_reducers);

        //job.setJarByClass(UserJarClass);
        job.setJobName("ListHomoMorphismStep1");

        job.setInputFormatClass(inputFormatClass);
        job.setOutputFormatClass(outputFormatClass);

        job.setMapperClass(UserFirstMapperClass);
        job.setReducerClass(UserFirstReducerClass);

        ClassLoader myloader = HomomorphismWrapperFor3rdThm.class.getClassLoader();
        String KeyClassName = job.getConfiguration().get(IndexKeyClassProperty);
        if (KeyClassName != null && KeyClassName.length() > 0) {
            Class KeyClass = myloader.loadClass(KeyClassName);
            job.setOutputKeyClass(KeyClass);
        } else {//default
            job.setOutputKeyClass(LongPair.class); //default LongPair
        }
        // job.setOutputValueClass(LongPair.class); //?? FoldingOutValClassProperty  FoldingInValClassProperty
        String OutputClassName = job.getConfiguration().get(FoldingOutValClassProperty);
        job.setOutputValueClass(myloader.loadClass(OutputClassName ));

        // maybe not necessary
        job.setMapOutputKeyClass(LongPair.class);
        String MapOutputClassName = job.getConfiguration().get(FoldingInValClassProperty);
        Class MapOutputValueClass = myloader.loadClass(MapOutputClassName);
        job.setMapOutputValueClass(MapOutputValueClass);

        job.setSortComparatorClass(TotalLongPairKeyComparator.class);
        job.setGroupingComparatorClass(GroupLongPairComparator.class);

        job.setPartitionerClass(LHPartitioner.class);
        //job.setCombinerClass(LHCominber.class); //results OK, but not efficient

        return job;

    }

    public Job initJobConf2(Job job,
            Class User2ndMapperClass,
            Class User2ndReducerClass,
            String[] args)
            throws IOException, ClassNotFoundException {

        ClassLoader myloader = HomomorphismWrapperFor3rdThm.class.getClassLoader();
        Class<? extends InputFormat> inputFormatClass =
                SequenceFileInputFormat.class;
        Class<? extends OutputFormat> outputFormatClass =
                TextOutputFormat.class;

        FileInputFormat.setInputPaths(job, args[1]);
        FileOutputFormat.setOutputPath(job, new Path(args[1]+ "_2p"));

        JobClient client = new JobClient(job.getConfiguration());
        ClusterStatus cluster = client.getClusterStatus();

        int num_reducers = (int) (cluster.getMaxReduceTasks() * 0.9);
        String sort_reduces = job.getConfiguration().get(REDUCES_PER_HOST);
        if (sort_reduces != null) {
            num_reducers = cluster.getTaskTrackers()
                    * Integer.parseInt(sort_reduces);
        }
        job.setNumReduceTasks(1);

        //job.setJarByClass(UserJarClass);
        job.setJobName("ListHomoMorphismStep2");

        job.setInputFormatClass(inputFormatClass);
        job.setOutputFormatClass(outputFormatClass);

        TextOutputFormat.setOutputPath(job, new Path(args[1] + "_final"));

        job.setJobName("ListHomoMorphismStep2");
        //   job.setJarByClass(UserJarClass);

        job.setMapperClass(User2ndMapperClass);
        job.setMapOutputKeyClass(LongPair.class);
      //  job.setMapOutputValueClass(LongWritable.class);  //??
        String MapOutputClassName = job.getConfiguration().get(FoldingInValClassProperty);
        Class MapOutputValueClass = Class.forName(MapOutputClassName);
        job.setMapOutputValueClass(MapOutputValueClass);

        job.setReducerClass(User2ndReducerClass);
        //    job.setCombinerClass(FirstReducer.class);

        job.setOutputKeyClass(NullWritable.class);
        String OutputClassName = job.getConfiguration().get(FoldingOutValClassProperty);
        job.setOutputValueClass(myloader.loadClass(OutputClassName ));
        // job.setOutputValueClass(LongPair.class);
        // job.setGroupingComparatorClass(GroupLongPairComparator.class);
        // job.setSortComparatorClass(KeyComparator.class);
//        job.setOutputKeyComparatorClass(TotalLongPairKeyComparator.class);
//        job.setOutputValueGroupingComparator(GroupLongPairComparator.class);
//        job.setPartitionerClass(SecondPartitioner.class);

        job.setSortComparatorClass(TotalLongPairKeyComparator.class);
        job.setGroupingComparatorClass(GroupLongPairComparator.class);

        job.setPartitionerClass(LHPartitioner.class);
        return job;
    }

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> <out>");
            System.exit(2);
        }
        // add user conf file
        Path userConfFile = new Path(args[2]);
        conf.addResource(userConfFile);
        int res = ToolRunner.run(conf, new HomomorphismWrapperFor3rdThm(), args);
        System.exit(res);

    }

    public static int HasParameter(String query, String[] pars) {
        boolean rst = false;
        int id = -1; //pars.length;
        for (String par : pars) {
            id++;
            if (par.equals(query)) {
                rst = true;
                //  id ++;
                break;
            }
        }
        if (!rst) {
            id = -1;
        }
        return id;
    }
}
